package com.es.rocket;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableConfigurationProperties(RocketmqProperties.class)
@ConditionalOnProperty(prefix = RocketmqProperties.PREFIX, value="namesrvAddr")
public class RocketConfig {
	
	@Autowired
	private RocketmqProperties properties;
	
	@Autowired
	private ApplicationEventPublisher publisher;
	
	/**
	 * https://www.jianshu.com/p/f57de40621a0
	 * 创建发送普通消息的生产者
	 * @return
	 * @throws MQClientException
	 */
	@Bean
	@ConditionalOnProperty(prefix = RocketmqProperties.PREFIX, value="producerInstanceName")
	public DefaultMQProducer defaultProducer() throws MQClientException{
		/**
		 * 创建一个producer
		 * producerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
		 * 因为服务器会回查这个Group下的任意一个Producer
		 */
		DefaultMQProducer producer = new DefaultMQProducer(properties.getProducerGroupName());
		producer.setNamesrvAddr(properties.getNamesrvAddr());
		producer.setInstanceName(properties.getProducerInstanceName());
		producer.setVipChannelEnabled(false);
		producer.setRetryTimesWhenSendFailed(10);
		/**
		 * producer对象在使用之前必须调用start初始化，初始化一次即可
		 * 注意：切记不可以在每次发送消息时，都调用start方法
		 */
		producer.start();
		return producer;
	}
	/**
	 * 创建发送事务消息的生产者
	 * @return
	 * @throws MQClientException
	 */
	@Bean
	@ConditionalOnProperty(prefix = RocketmqProperties.PREFIX, value = "producerTranInstanceName")
	public TransactionMQProducer transactionProducer() throws MQClientException{
		TransactionMQProducer producer = new TransactionMQProducer(properties.getTransactionProducerGroupName());
		producer.setNamesrvAddr(properties.getNamesrvAddr());
		producer.setInstanceName(properties.getProducerTranInstanceName());
		//事务回查最小并发数
		producer.setCheckThreadPoolMinSize(2);
		//事务回查最大并发数
		producer.setCheckThreadPoolMaxSize(2);
		//队列数
		producer.setCheckRequestHoldMax(2000);
		//由于社区版本的服务器阉割调了消息回查的功能，所以这个地方没有意义
		//TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();
		//producer.setTransactionCheckListener(transactionCheckListener);
		//初始化producer
		producer.start();
		return producer;
	}
	
	@Bean
	public DefaultMQPushConsumer pushConsumer() throws MQClientException{
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(properties.getConsumerGroupName());
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		consumer.setNamesrvAddr(properties.getNamesrvAddr());
		consumer.setInstanceName(properties.getConsumerInstanceName());
		if (properties.isConsumerBroadcasting()) {
			consumer.setMessageModel(MessageModel.BROADCASTING);
		}
		//设置批量消费，以提消费吞吐量，默认1
		consumer.setConsumeMessageBatchMaxSize(
				properties.getConsumerBatchMaxSize() ==0 ? 1: properties.getConsumerBatchMaxSize());
		/**
		 * 订阅指定topic下的tags
		 */
		List<String> subscribeList = properties.getSubscribe();
		if (subscribeList != null && !subscribeList.isEmpty()) {
			for (String subscribe : subscribeList) {
				consumer.subscribe(subscribe.split(":")[0], subscribe.split(":")[1]);
			}
		}
		
		if (properties.isEnableOrderConsumer()) {
			//顺序消费
			orderConsumer(consumer);
		}else {
			consumer(consumer);
		}
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				try {
					/**
					 * 延迟5秒再启动，主要是等待spring事件监听相关程序初始化完成，
					 * 否则会出现对rocketmq的消息进行消费后立即发消息到大事件，
					 * 然而此事件的监听程序还未初始化，从而造成消息的丢失
					 */
					Thread.sleep(5000);
					//Consumer对象在使用之前必须要调用start初始化，初始化一次即可
					consumer.start();
				} catch (Exception e) {
					// TODO: handle exception
				}
				
			}
		}).start();
		return consumer;
	}
	private void consumer(DefaultMQPushConsumer consumer) {
		//注册消费监听
		consumer.registerMessageListener(new MessageListenerConcurrently() {
			
			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
					ConsumeConcurrentlyContext context) {
				try {
					//msgs = filter(msgs);
					if (msgs == null || msgs.isEmpty()) {
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
					Map<String, Object> map = new HashMap<String, Object>();
					map.put("msgs", msgs);
					map.put("consumer", consumer);
					publisher.publishEvent(new RocketmqEvent(map));
				} catch (Exception e) {
					e.printStackTrace();
					return ConsumeConcurrentlyStatus.RECONSUME_LATER;
				}
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			
			}
		});
	}
	private void orderConsumer(DefaultMQPushConsumer consumer) {
		consumer.registerMessageListener(new MessageListenerOrderly(){

			@Override
			public ConsumeOrderlyStatus consumeMessage(
					List<MessageExt> msgs, ConsumeOrderlyContext context) {
				try {
					context.setAutoCommit(true);
					//msgs = filter(msgs);
					if (msgs == null || msgs.isEmpty()) {
						return ConsumeOrderlyStatus.SUCCESS;
					}
					Map<String, Object> map = new HashMap<String, Object>();
					map.put("msgs", msgs);
					map.put("consumer", consumer);
					publisher.publishEvent(new RocketmqEvent(map));
				} catch (Exception e) {
					e.printStackTrace();
					return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
				}
				return ConsumeOrderlyStatus.SUCCESS;
			}
		});
	}
	
	
	private List<MessageExt> filter(List<MessageExt> msgs) {
		// TODO Auto-generated method stub
		return null;
	}
}
